/*
 *  Queen Mary University of London MOM System
 *  
 *  Copyright Queen Mary University of London
 *  
 *  Authors:
 *  Bob Chew (bob.chew@elec.qmul.ac.uk)
 *  Beatriz Viñal Murciano (beavimu@gmail.com)
 *  John Bigham (john.bigham@elec.qmul.ac.uk)
 *  
 *  
 *  Portions of this software uses Apache Qpid
 *    http://qpid.apache.org
 *  
 *  Apache Qpid license:
 * --------------------------------------------------------------
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 * -------------------------------------------------------------
 * 
 */
package users;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
import java.util.UUID;

import javax.jms.JMSException;
import javax.jms.TopicSubscriber;

import org.apache.qpid.AMQException;
import org.apache.qpid.client.message.JMSTextMessage;
import org.apache.qpid.url.URLSyntaxException;

/**
 * Subscriber to a Qpid broker.
 *
 * What to do with the received messages is application dependent and must be
 * implemented at application level.
 */
public class Subscriber extends User {

    /** List of the subscriptions of the subscriber. */
    private Set<Subscription> subscriptions = new HashSet<Subscription>();
    //TODO: should we allow more than one subscription to a topic?
    //(i.e. subscriptions with different message selectors)
    /** Number of subscriptions. */
    private int numSubs = subscriptions.size();
    /** Subscriptions mapped to their mirror subscribers. */
    private HashMap<Subscription, MirrorSubscriber> subsMirrors
            = new HashMap<Subscription, MirrorSubscriber>();
    /** Message listener to messages from application clients. */
    private SubscriberMessageListener listener
            = new SubscriberMessageListener(this);

    /**
     * Getter for the variable 'subscriptions'.
     * @return variable 'subscriptions'
     */
    public Set<Subscription> getSubscriptions() {
        return subscriptions;
    }
    /**
     * Setter for the variable 'subscriptions'.
     * @param subscriptions  new value of the variable 'subscriptions'
     */
    public void setSubscriptions(final Set<Subscription> subscriptions) {
        this.subscriptions = subscriptions;
    }
    /**
     * Getter for the variable 'subsMirrors'.
     * @return variable 'subsMirrors'
     */
    public HashMap<Subscription, MirrorSubscriber> getSubsMirrors() {
        return subsMirrors;
    }
    /**
     * Setter for the variable 'subsMirrors'.
     * @param subsMirrors  new value of the variable 'subsMirrors'
     */
    public void setSubsMirrors(final HashMap<Subscription,
                               MirrorSubscriber> subsMirrors) {
        this.subsMirrors = subsMirrors;
    }
    /**
     * Getter for the variable 'numSubs'.
     * @return variable 'numSubs'
     */
    public int getNumSubs() {
        return numSubs;
    }
    /**
     * Setter for the variable 'numSubs'.
     * @param numSubs  new value of the variable 'numSubs'
     */
    public void setNumSubs(final int numSubs) {
        this.numSubs = numSubs;
    }
    /**
     * Getter for the variable 'listener'.
     * @return variable 'listener'
     */
    public SubscriberMessageListener getListener() {
        return listener;
    }
    /**
     * Setter for the variable 'listener'.
     * @param listener  new value of the variable 'listener'
     */
    public void setListener(final SubscriberMessageListener listener) {
        this.listener = listener;
    }


    /**
     * Empty constructor with no parameters.
     */
    public Subscriber() {
        //after retrieving a subscriber from a database, its scAssigTopic,
        //scUpdatesTopic and omUpMsgListener MUST be set manually
    }

    /**
     * Constructor for new subscribers.
     *
     * @param username  username to use to connect to the broker
     * @param password  password to use to connect to the broker
     */
    public Subscriber(final String username, final String password) {

        super(username, password);

        System.out.println("Subscriber " + getClientId() + " started");

        //get the name of the broker to publish to
        requestBroker();

    }

    /**
     * Get the subscription with a topic name.
     *
     * @param topicName  topic of the subscription to get
     * @return  subscription with topic topicName
     */
    public Subscription getSubscription(final String topicName) {
        for (Subscription sub : getSubscriptions()) {
            if (sub.getTopic().equals(topicName)) {
                return sub;
            }
        }
        return null;
    }

    /**
     * Get the connection in which a subscription is.
     *
     * @param sub  subscription to find
     * @return  connection with the subscription sub
     *          or null if it doesn't exist
     */
    private SubscriberConnection getConnection(final Subscription sub) {
        for (UserConnection con : getConnections()) {
            if (((SubscriberConnection) con).getSubscriptions().keySet()
                        .contains(sub)) {
                return (SubscriberConnection) con;
            }
        }
        return null;
    }

    /**
     * Add a subscription (topic name and message selector) to the list of
     * subscriptions, create a subscriber for it and start listening to its
     * messages.
     *
     * @param topicName  name of the topic to subscribe to
     * @param msgSelector  message selector for that topic
     * @return  the subscription added or null if it already existed or if
     *             there is any problem
     */
    public Subscription addSubscription(final String topicName,
                                               final String msgSelector) {
        //if a subscription to that topic doesn't already exist
        if (getSubscription(topicName) == null) {
            //get a random subscription ID
            String subscriptionId = UUID.randomUUID().toString();
            //create the subscription
            Subscription subscription = new Subscription(
                    subscriptionId, topicName, msgSelector);
            //add the subscription to the lists of subscriptions
            getSubscriptions().add(subscription);
            //update numSubs
            setNumSubs(getSubscriptions().size());

            //ask the overlay manager what broker to use
            try {
                //send the message to the overlay manager
                String msgText = "broker request:subscriber:" + getClientId()
                                 + ":" + subscription.getTopic();
                getOmConnection().sendMessage(getOmAssigPub(), msgText);

                System.out.println("Subscriber " + getClientId()
                        + " waiting for a broker to subscribe: " + msgText);

                //wait until a message is received
                JMSTextMessage recMsg
                        = (JMSTextMessage) getOmAssigSub().receive();

                //analyse the text of the received message
                String text = recMsg.getText();
                String[] parts = text.split(":");
                if (parts.length == NUM_PARTS_ASSIG_MSG) {

                    if (parts[MB_INDEX_ASSIG].equals("null")) {
                        throw new Exception("No mirror broker assigned");
                    }
                    //create mirror subscriber in mirror broker
                    createMSForConn(subscription, parts[MB_INDEX_ASSIG],
                            Integer.parseInt(parts[MP_INDEX_ASSIG]));

                    if (parts[PB_INDEX_ASSIG].equals("null")) {
                        throw new Exception("No primary broker assigned");
                    }
                    //add the subscription to connection with the primary broker
                    SubscriberConnection sCon = addSubToConn(subscription,
                            parts[PB_INDEX_ASSIG],
                            Integer.parseInt(parts[PP_INDEX_ASSIG]));

                    System.out.println("Subscriber " + getClientId()
                            + " has been assigned broker " + sCon.getBroker()
                            + " and port " + sCon.getPort());

                    return subscription;
                } else {
                    throw new Exception("Error assigning brokers: " + text);
                }

            } catch (Exception e) {
                System.err.println(e.getMessage());
            }

        }
        return null;
    }

    /**
     * Unsubscribe the subscriber from the topic of the given name.
     *
     * @param topicName  name of the topic to unsubscribe from
     * @throws JMSException  if there is any problem
     */
    public void removeSubscription(final String topicName)
            throws JMSException {
        Subscription sub = getSubscription(topicName);
        //find the SubscriberConnection in which the subscription is
        SubscriberConnection con = getConnection(sub);
        //delete the subscription in the SubscriberConnection
        con.removeSubscription(sub);
        //find the mirror subscriber
        MirrorSubscriber ms = getSubsMirrors().get(sub);
        //delete the subscription in the mirror connection
        ms.removeSubscription();
        //delete the subscription and the mirror subscriber
        getSubscriptions().remove(sub);
        getSubsMirrors().remove(sub);
        //update numSubs
        setNumSubs(getSubscriptions().size());
    }

    /**
     * Add a subscription to the connection with the given broker.
     * Create the connection if it does not exist.
     *
     * @param subscription  to add
     * @param broker  IP address of the broker
     * @param port  port of the broker
     * @return  connection in which the subscription has been added
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     * @throws URLSyntaxException  if there is any problem
     */
    private SubscriberConnection addSubToConn(final Subscription subscription,
            final String broker, final int port)
            throws JMSException, URLSyntaxException, AMQException {

        //get a connection with the broker
        SubscriberConnection conn
                = (SubscriberConnection) getConnForBroker(broker);

        if (conn == null) {
            conn = new SubscriberConnection(this, broker, port,
                                            getMOM_EXCHANGE());
            conn.initialize();
            //add the primary connection to the list of connections
            getConnections().add(conn);
        }
        //add the subscription to the connection
        conn.addSubscription(subscription);

        return conn;

    }

    /**
     * Create a mirror reader that uses the connection with the given broker.
     * Create the connection if it does not exist.
     *
     * @param subscription  to create a mirror subscriber for
     * @param broker  IP address of the broker
     * @param port  port of the broker
     * @return  connection in which the mirror subscriber has been created
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     * @throws URLSyntaxException  if there is any problem
     */
    private SubscriberConnection createMSForConn(
            final Subscription subscription, final String broker,
            final int port)
            throws JMSException, URLSyntaxException, AMQException {

        //get a connection with the broker
        SubscriberConnection conn
                = (SubscriberConnection) getConnForBroker(broker);
        if (conn == null) {
            conn = new SubscriberConnection(this, broker, port,
                                            getMOM_EXCHANGE());
            conn.initialize();
            //add the primary connection to the list of connections
            getConnections().add(conn);
        }
        //create the mirror subscriber
        MirrorSubscriber ms = new MirrorSubscriber(
                this, conn.getConnection(), subscription);
        //store it
        getSubsMirrors().put(subscription, ms);

        return conn;

    }

    /**
     * When a user is created, it asks the overlay manager what brokers it has
     * to use sending the overlay manager a message.
     *
     * The format of the message is:
     *
     * 'broker request:subscriber:<clientId>:<topicName>:<topicName>:...'
     *
     * The format of the received message is:
     *
     * '<topicName>:<primaryBroker>:<primaryPort>:<mirrorBroker>:<mirrorPort>:...'
     */
    @Override
    public void requestBroker() {

        try {
            //initialize connection with the overlay manager
            getOmConnection().initialize();

            //create the subscriber to messages about subscriptions from the
            //overlay manager
            setOmAssigSub(getOmConnection().getSubscriberForTopic(
                    getOmAssigTopic()));

            //create the publisher to send messages to the overlay manager
            setOmAssigPub(
                    getOmConnection().getPublisherForTopic(getOM_TOPIC()));

            //request broker only if there are any subscriptions
            if (!getSubscriptions().isEmpty()) {
                //send the message to the overlay manager
                String msgText = "broker request:subscriber:" + getClientId();
                for (Subscription sub : getSubscriptions()) {
                    msgText = msgText + ":" + sub.getTopic();
                }
                getOmConnection().sendMessage(getOmAssigPub(), msgText);
                //wait until a message is received
                System.out.println("Subscriber " + getClientId() + " waiting "
                        + "for being assigned a broker: " + msgText);
                JMSTextMessage recMsg = (JMSTextMessage) getOmAssigSub()
                        .receive();
                //analyse the text of the received message
                String text = recMsg.getText();
                String[] parts = text.split(":");
                if ((parts.length % NUM_PARTS_ASSIG_MSG) == 0) {
                    for (int i = 0;
                            i < (parts.length - (NUM_PARTS_ASSIG_MSG - 1));
                            i += NUM_PARTS_ASSIG_MSG) {
                        String topic = parts[i]; //topic

                        //get the subscription of that topic
                        Subscription subscription = getSubscription(topic);

                        if (parts[i + MB_INDEX_ASSIG].equals("null")) {
                            throw new Exception("No mirror broker assigned");
                        }
                        //create mirror subscriber in mirror broker
                        createMSForConn(subscription,
                                parts[i + MB_INDEX_ASSIG],
                                Integer.parseInt(parts[i + MP_INDEX_ASSIG]));

                        if (parts[i + PB_INDEX_ASSIG].equals("null")) {
                            throw new Exception("No primary broker assigned");
                        }
                        //add the subscription to connection with primary broker
                        SubscriberConnection sCon = addSubToConn(subscription,
                                parts[i + PB_INDEX_ASSIG],
                                Integer.parseInt(parts[i + PP_INDEX_ASSIG]));

                        System.out.println("Subscriber " + getClientId()
                                + " has been assigned broker "
                                + sCon.getBroker() + " and port "
                                + sCon.getPort() + " for topic " + topic);
                    }
                } else {
                    throw new Exception("Error assigning brokers: " + text);
                }

            }

            //create a subscriber for updates from the overlay manager
            TopicSubscriber subscriber = getOmConnection()
                    .getSubscriberForTopic(getOmUpdatesTopic());
            //set a message listener for updates
            subscriber.setMessageListener(getOmUpMsgListener());

        } catch (Exception e) {
            System.err.println("It hasn't been possible to assign a broker");
            e.printStackTrace();
        }
    }

    /**
     * Change the primary broker of a subscriber for a topic.
     * The subscription of the given topic is moved to the broker and port of
     * its mirror subscriber and its mirror subscriber is moved to the given
     * broker and port.
     *
     * @param topic  topic to change
     * @param newBroker  new mirror broker to use
     * @param newPort  new mirror port to use
     * @throws URLSyntaxException  if there is any problem
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     */
    protected void changePrimary(final String topic, final String newBroker,
            final int newPort)
            throws URLSyntaxException, JMSException, AMQException, Exception {

        //subscription for the topic
        Subscription sub = getSubscription(topic);

        if (sub != null) {
            //mirror subscriber for the topic
            MirrorSubscriber ms = getSubsMirrors().get(sub);

            //get mirror broker and mirror port (new primaries)
            String primaryBroker = ms.getHost();
            int primaryPort = ms.getPort();

            //connection of the subscription
            SubscriberConnection oldCon = getConnection(sub);
            try {
                //unsubscribe subscription in old broker
                oldCon.removeSubscription(sub);
                //if the connection is closed, remove it
                if (!oldCon.isConnected()) {
                    disconnect(oldCon);
                }
            } catch (Exception e) {
                //connection didn't exist or was already closed
                disconnect(oldCon);
            }

            //close old mirror subscriber
            ms.close();
            getSubsMirrors().remove(sub);

            if (newBroker.equals("null")) {
                throw new Exception("No mirror broker assigned");
            }
            //create new mirror subscriber in new mirror broker
            createMSForConn(sub, newBroker, newPort);

            //add subscription to connection with new primary broker
            SubscriberConnection sCon
                    = addSubToConn(sub, primaryBroker, primaryPort);

            System.out.println("Subscriber " + getClientId() + " has moved the"
                    + " subscription to " + sub.getTopic() + " to broker "
                    + sCon.getBroker() + " and port " + sCon.getPort());
        }

    }

    /**
     * Change the mirror broker of a subscriber for a topic.
     * The mirror subscriber of the given topic is closed and a replacement is
     * created in the given broker at the given port.
     *
     * @param topic  topic to change
     * @param newBroker  new mirror broker to use
     * @param newPort  new mirror port to use
     * @throws URLSyntaxException  if there is any problem
     * @throws JMSException  if there is any problem
     * @throws AMQException  if there is any problem
     */
    protected void changeMirror(final String topic, final String newBroker,
            final int newPort)
            throws URLSyntaxException, JMSException, AMQException, Exception {

        //subscription for the topic
        Subscription sub = getSubscription(topic);

        if (sub != null) {
            //mirror subscriber for the topic
            MirrorSubscriber ms = getSubsMirrors().get(sub);

            //remove the old connection from the list
            String oldBroker = ms.getHost();
            UserConnection oldCon = getConnForBroker(oldBroker);

            try {
                //unsubscribe subscription in old broker
                ms.removeSubscription();
                //if the connection is closed, remove it
                if (!oldCon.isConnected()) {
                    disconnect(oldCon);
                }
            } catch (Exception e) {
                //connection didn't exist or was already closed
                disconnect(oldCon);
            }

            //close old mirror subscriber
            ms.close();
            getSubsMirrors().remove(sub);

            if (newBroker.equals("null")) {
                throw new Exception("No mirror broker assigned");
            }
            //create new mirror subscriber in new mirror broker
            SubscriberConnection msCon
                    = createMSForConn(sub, newBroker, newPort);

            System.out.println("Subscriber " + getClientId() + " has moved the"
                    + " mirror subscriber to " + sub.getTopic() + " to broker "
                    + msCon.getBroker() + " and port " + msCon.getPort());
        }

    }

}
